package com.fanxl.flink.project.udf;

import com.fanxl.flink.project.domain.EventCategoryProductCount;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/3/21 13:07
 */
public class TopNWindowFunction implements WindowFunction<Long, EventCategoryProductCount, Tuple3<String, String, String>, TimeWindow> {

    @Override
    public void apply(Tuple3<String, String, String> value, TimeWindow window, Iterable<Long> input, Collector<EventCategoryProductCount> out) throws Exception {
        Long count = input.iterator().next();
        out.collect(new EventCategoryProductCount(value.f0, value.f1, value.f2, count, window.getStart(), window.getEnd()));
    }
}
